const { Readable, Writable, Transform, pipeline } = require('stream');

// A simple readable stream that emits data
class MyReadable extends Readable {
  constructor(options) {
    super(options);
    this.data = ["Node.js", "Streams", "are", "powerful"];
    this.index = 0;
  }

  _read(size) {
    if (this.index < this.data.length) {
      this.push(this.data[this.index]);
      this.index += 1;
    } else {
      this.push(null); // No more data
    }
  }
}

// A simple writable stream that consumes data
class MyWritable extends Writable {
  _write(chunk, encoding, callback) {
    console.log(chunk.toString());
    callback();
  }
}

// A simple transform stream that modifies data
class MyTransform extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
}

const readable = new MyReadable();
const writable = new MyWritable();
const transform = new MyTransform();

// Using pipeline to handle stream events and errors
pipeline(
  readable,
  transform,
  writable,
  (err) => {
    if (err) {
      console.error('Pipeline failed.', err);
    } else {
      console.log('Pipeline succeeded.');
    }
  }
);